# Copyright (c) 2018 Presto Labs Pte. Ltd.
# Author: jaewon

import datetime

from absl import app, flags
from tornado.ioloop import IOLoop

from coin.util.queue.reader import AsyncKafkaReader

from coin.util.queue.config import KafkaConfig

FLAGS = flags.FLAGS

flags.DEFINE_string('kafka_config_filename', None, 'kafka config')

flags.DEFINE_string('topic', 'example_topic', '')

flags.DEFINE_integer('count', 10000, '')


async def loop():
  kafka_config = KafkaConfig.from_cmd_config(FLAGS.kafka_config_filename)
  reader = AsyncKafkaReader(FLAGS.topic.split(','),
                            kafka_config=kafka_config,
                            timestamp_from=datetime.datetime.now() - datetime.timedelta(minutes=10))
  await reader.open()
  try:
    for counter in range(FLAGS.count):
      records = await reader.read()
      print(records)
  finally:
    await reader.close()

  ioloop = IOLoop.current()
  ioloop.stop()


def main(argv):
  ioloop = IOLoop.current()
  ioloop.add_callback(loop)
  try:
    ioloop.start()
  except KeyboardInterrupt:
    print()
  ioloop.stop()


if __name__ == "__main__":
  app.run(main)
